package com.dys.flink.wc2;

import com.dys.flink.wc.util.JsonUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Socket 实时订单流，按 product_id 累加销售额和订单数（KeyedState 示例）
 * 运行前：nc -lk 9999
 */
public class StatefulOrderJob {

    public static void main(String[] args) throws Exception {
        try (StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment()) {
            env.setParallelism(1);

            // 1. socket 源
            DataStream<String> text = env.socketTextStream("localhost", 9999);

            // 2. 解析
            DataStream<Order> orders = text
                    .filter(l -> l != null && !l.trim().isEmpty())
                    .map(l -> JsonUtils.fromJson(l, Order.class))
                    .returns(Order.class);

            // 3. 按 product_id 分组 + 状态处理
            orders.keyBy(o -> o.product_id)
                    .process(new AccumulateSalesProc())
                    .print(); // 每来一条就输出当前累计值

            env.execute("StatefulOrderJob");
        }
    }

    // ---------------- KeyedProcessFunction 状态逻辑 ----------------
    public static class AccumulateSalesProc extends KeyedProcessFunction<String, Order, Stat> {

        // 状态：当前商品的 (订单数, 总销售额)
        private ValueState<Stat> state;

        @Override
        public void open(Configuration parameters) {
            ValueStateDescriptor<Stat> descriptor =
                    new ValueStateDescriptor<>("sale-stat", TypeInformation.of(Stat.class));
            state = this.getRuntimeContext().getState(descriptor);
        }

        @Override
        public void processElement(Order order, Context ctx, Collector<Stat> out) throws Exception {
            Stat current = state.value();
            if (current == null) {
                current = new Stat(order.product_id, 0L, 0.0);
            }
            current.count += 1;
            current.totalSales += order.product_price;
            state.update(current);
            out.collect(current); // 实时输出
        }
    }

    // ---------------- POJO ----------------
    public static class Order {
        public long id;
        public String product_name;
        public String product_id;
        public double product_price;
    }

    public static class Stat {
        public String productId;
        public long count;
        public double totalSales;

        public Stat() {}

        public Stat(String productId, long count, double totalSales) {
            this.productId = productId;
            this.count = count;
            this.totalSales = totalSales;
        }

        @Override
        public String toString() {
            return String.format("Stat{productId='%s', count=%d, totalSales=%.2f}", productId, count, totalSales);
        }
    }
}